package com.ada.spark.rddoperator

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 作用：类似于mapPartitions，但func带有一个整数参数表示分片的索引值，因此在类型为T的RDD上运行时，func的函数类型必须是(Int, Interator[T]) => Iterator[U]；
  */
object Spark04_mapPartitionsWithIndex {

    def main(args: Array[String]): Unit = {
        //创建SparkConf
        val conf = new SparkConf().setAppName("Spark04_mapPartitionsWithIndex").setMaster("local[*]")
        //创建Spark上下文对象
        val sc = new SparkContext(conf)

        val listRDD: RDD[Int] = sc.makeRDD(1 to 10)

        //类似于mapPartitions，但func带有一个整数参数表示分片的索引值，因此在类型为T的RDD上运行时，func的函数类型必须是(Int, Interator[T]) => Iterator[U]
        val mapPartitionsWithIndexRDD: RDD[(Int, String)] = listRDD.mapPartitionsWithIndex {
            case (num, datas) => {
                datas.map((_, "分区号" + num))
            }
        }

        println(mapPartitionsWithIndexRDD.collect().mkString(","))
        //(1,分区号0),(2,分区号0),(3,分区号1),(4,分区号1),(5,分区号1),(6,分区号2),(7,分区号2),(8,分区号3),(9,分区号3),(10,分区号3)
    }

}
